package com.polarrose.amazon.sqs.impl;

import java.net.MalformedURLException;
import java.net.URL;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.commons.httpclient.HostConfiguration;
import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.HttpStatus;
import org.apache.commons.httpclient.MultiThreadedHttpConnectionManager;
import org.apache.commons.httpclient.NameValuePair;
import org.apache.commons.httpclient.methods.GetMethod;

import com.polarrose.amazon.aws.AuthenticationUtils;
import com.polarrose.amazon.aws.AwsAccount;
import com.polarrose.amazon.sqs.QueueAttribute;
import com.polarrose.amazon.sqs.SqsCoreService;
import com.polarrose.amazon.sqs.SqsException;
import com.polarrose.amazon.sqs.SqsMessage;
import com.polarrose.amazon.sqs.SqsQueue;
import com.polarrose.amazon.sqs.SqsUtils;

public class SqsCoreServiceImpl implements SqsCoreService {

	public final static String SQS_API_VERSION = "2008-01-01";
	public final static String SQS_SERVICE_HOST = "http://queue.amazonaws.com";

	protected HttpClient client;
	protected AwsAccount account;

	public SqsCoreServiceImpl() {
		init();
	}

	public SqsCoreServiceImpl(AwsAccount account) {
		this();
		this.account = account;
	}

	private void init() {
		MultiThreadedHttpConnectionManager connectionManager = new MultiThreadedHttpConnectionManager();
		connectionManager.getParams().setConnectionTimeout(4000);
        connectionManager.getParams().setDefaultMaxConnectionsPerHost(250);
        connectionManager.getParams().setMaxConnectionsPerHost(HostConfiguration.ANY_HOST_CONFIGURATION, 500);
        connectionManager.getParams().setMaxTotalConnections(2000);
        connectionManager.getParams().setSoTimeout(15000);
		client = new HttpClient(connectionManager);
	}

	public void setAccount(AwsAccount account) {
		this.account = account;
	}

	public SqsQueue createQueue(String queueName) throws SqsException {
		return createQueue(queueName, 30);
	}

	public SqsQueue createQueue(String queueName, int visibilityTimeout) throws SqsException {
		SqsUtils.validateQueueName(queueName);
		Map<String, String> parameters = new HashMap<String, String>();
		parameters.put("QueueName", queueName);
		parameters.put("DefaultVisibilityTimeout", String.valueOf(visibilityTimeout));

		CreateQueueResponse response = executeServiceRequest(CreateQueueResponse.class, account, "CreateQueue", parameters);

		return new SqsQueue(response.getQueueUrl());
	}

	public void deleteQueue(SqsQueue queue) throws SqsException {
		executeQueueRequest(DeleteQueueResponse.class, account, queue, "DeleteQueue", null);
	}

	public Map<QueueAttribute, Object> getQueueAttributes(SqsQueue queue, QueueAttribute attribute) {
		Map<String, String> parameters = new HashMap<String, String>();
		if (attribute == null) {
			parameters.put("AttributeName", QueueAttribute.All.name());
		} else {
			parameters.put("AttributeName", attribute.name());
		}
		GetQueueAttributesResponse response = executeQueueRequest(GetQueueAttributesResponse.class, account, queue, "GetQueueAttributes", parameters);

		Map<QueueAttribute, Object> attributes = new HashMap<QueueAttribute, Object>();

		Map<String, String> returnedAttributes = response.getAttributes();
		for (Entry<String, String> entry : returnedAttributes.entrySet()) {
			if (entry.getKey().equals(QueueAttribute.ApproximateNumberOfMessages.name())) {
				String value = entry.getValue();
				if (value == null) {
					attributes.put(QueueAttribute.ApproximateNumberOfMessages, 0);
				} else {
					attributes.put(QueueAttribute.ApproximateNumberOfMessages, Integer.valueOf(value));
				}
			} else if (entry.getKey().equals(QueueAttribute.VisibilityTimeout.name())) {
				String value = entry.getValue();
				if (value == null) {
					attributes.put(QueueAttribute.VisibilityTimeout, 0);
				} else {
					attributes.put(QueueAttribute.VisibilityTimeout, Integer.valueOf(value));
				}
			}
		}
		return attributes;
	}

	public List<SqsQueue> listQueues() throws SqsException {
		return listQueues(null);
	}

	public List<SqsQueue> listQueues(String queueNamePrefix) throws SqsException {
		Map<String, String> parameters = new HashMap<String, String>();
		if (queueNamePrefix != null) {
			parameters.put("QueueNamePrefix", queueNamePrefix);
		}
		ListQueuesResponse response = executeServiceRequest(ListQueuesResponse.class, account, "ListQueues", parameters);
		return response.getQueues();
	}

	public void setQueueAttributes(SqsQueue queue, QueueAttribute attribute, Object value) {
		Map<String, String> parameters = new HashMap<String, String>();
		parameters.put("Attribute.Name", attribute.name());
		parameters.put("Attribute.Value", String.valueOf(value));

		executeQueueRequest(SetQueueAttributesResponse.class, account, queue, "SetQueueAttributes", parameters);
	}

	public SqsMessage sendMessage(SqsQueue queue, String body) {
		SqsUtils.validateBody(body);

		Map<String, String> parameters = new HashMap<String, String>();
		parameters.put("MessageBody", body);

		SendMessageResponse response = executeQueueRequest(SendMessageResponse.class, account, queue, "SendMessage", parameters);

		return new SqsMessage(response.getMessageId(), body, response.getMessageMd5());
	}

	public List<SqsMessage> receiveMessage(SqsQueue queue, int visibilityTimeout, int maxNumberOfMessages) {
		SqsUtils.validateVisibilityTimeout(visibilityTimeout);
		SqsUtils.validateNumberOfMessages(maxNumberOfMessages);

		Map<String, String> parameters = new HashMap<String, String>();
		parameters.put("MaxNumberOfMessages", Integer.toString(maxNumberOfMessages));
		parameters.put("VisibilityTimeout", Integer.toString(visibilityTimeout));

		ReceiveMessagesResponse response = executeQueueRequest(ReceiveMessagesResponse.class, account, queue, "ReceiveMessage", parameters);

		return response.getMessages();
	}

	public void deleteMessage(SqsQueue queue, String receiptHandle) {
		Map<String, String> parameters = new HashMap<String, String>();
		parameters.put("ReceiptHandle", receiptHandle);

		executeQueueRequest(DeleteMessageResponse.class, account, queue, "DeleteMessage", parameters);
	}

	private <T extends AbstractResponse> T executeQueueRequest(Class<T> responseParserClass, AwsAccount account, SqsQueue queue, String action, Map<String, String> parameters) throws SqsException {
		return executeRequest(responseParserClass, account, queue.getUrl(), action, parameters);
	}

	private <T extends AbstractResponse> T executeServiceRequest(Class<T> responseParserClass, AwsAccount account, String action, Map<String, String> parameters) throws SqsException {
		try {
			URL serviceUrl = new URL(SQS_SERVICE_HOST);
			return executeRequest(responseParserClass, account, serviceUrl, action, parameters);
		} catch (MalformedURLException e) {
			throw new SqsException("Cannot construct service url: " + e.getMessage());
		}
	}

	private <T extends AbstractResponse> T executeRequest(Class<T> responseParserClass, AwsAccount account, URL serviceEndpoint, String action, Map<String, String> parameters) throws SqsException {
		GetMethod method = null;

		try {
			// Create a method
			method = new GetMethod(serviceEndpoint.toExternalForm());

			// Build the parameter list, add the required ones
			Map<String, String> queryParameters = new HashMap<String, String>();
			if (parameters != null) {
				queryParameters.putAll(parameters);
			}

			queryParameters.put("Action", action);
			queryParameters.put("Version", SQS_API_VERSION);
			queryParameters.put("AWSAccessKeyId", account.getAccessKeyId());
			queryParameters.put("Timestamp", createTimestamp());
			queryParameters.put("SignatureVersion", "1");

			// Convert it to what the HTTP client wants

			NameValuePair[] nameValuePairs = new NameValuePair[queryParameters.size() + 1];

			int i = 0;
			for (Entry<String, String> queryParameter : queryParameters.entrySet()) {
				nameValuePairs[i] = new NameValuePair(queryParameter.getKey(), queryParameter.getValue());
				i++;
			}

			// Finally add the signature
			nameValuePairs[nameValuePairs.length - 1] = new NameValuePair("Signature", AuthenticationUtils.createSignature(account, generateSignatureData(queryParameters)));

			method.setQueryString(nameValuePairs);

			// Execute the request
			int responseCode = client.executeMethod(method);
			if (responseCode == HttpStatus.SC_OK) {
				T responseParser = responseParserClass.newInstance();
				responseParser.parse(method.getResponseBodyAsStream());

				return responseParser;
			} else if (responseCode == HttpStatus.SC_BAD_REQUEST) {

				ErrorResponse response = new ErrorResponse();
				response.parse(method.getResponseBodyAsStream());
				throw new SqsException("SQS Call Failed: " + response.getSqsError(), response.getSqsError());
			} else {
				throw new SqsException("SQS Service returned unknown response: code=" + responseCode + " message=" + method.getStatusText());
			}
		}

		catch (Exception e) {
			throw new SqsException("Cannot execute command: " + e.getMessage(), e);
		} finally {
			if (method != null) {
				method.releaseConnection();
			}
		}
	}

	private static final SimpleDateFormat timestampDateformat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ssZ");

	private String createTimestamp() {
		return timestampDateformat.format(new Date());
	}

	private String generateSignatureData(Map<String, String> parameters) {
		// Create a sorted list of the query parameter names

		List<String> sortedNames = new ArrayList<String>();
		for (String name : parameters.keySet()) {
			sortedNames.add(name);
		}

		Collections.sort(sortedNames, String.CASE_INSENSITIVE_ORDER);

		// Concatenate the sorted name/value pairs to form the signature data

		StringBuffer signatureData = new StringBuffer();

		for (String name : sortedNames) {
			signatureData.append(name);
			signatureData.append(parameters.get(name));
		}

		return signatureData.toString();
	}
}
